package com.monitor.web.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.Optional;

/**
 * @Author tanlian
 * @Date 2019/1/15 13:51
 * @Desc
 */
@Service
@Slf4j
public class ScheduledService {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Scheduled(cron = "0/5 * * * * *")
    public void test(){
        ListenableFuture future = kafkaTemplate.send("COM-TEST", "tanlian" );
        future.addCallback(o -> System.out.println("send-消息发送成功：" + "tanlian"), throwable -> System.out.println("消息发送失败：" + "tanlian"));
    }


    @KafkaListener(topicPattern = "COM-TEST",groupId="COM-TEST-ACTIVE")
    public void listenOne(ConsumerRecord<String, String> record) {
        Optional<String> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            log.info("topic : {}, mes : {}", record.topic(), kafkaMessage.get());
        }
    }

}
